package com.shanks.framework.mq.service.impl;

import com.shanks.framework.mq.enums.MQZone;
import com.shanks.framework.mq.enums.MessageRecordStatusEnum;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.model.MessageRecordPO;
import com.shanks.framework.mq.producer.BaseMQProducer;
import com.shanks.framework.mq.repository.MessageRecordRepository;
import com.shanks.framework.mq.service.MessagePushService;
import com.shanks.framework.mq.service.ProducerService;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.Objects;

@Service
public class MessagePushServiceImpl implements MessagePushService {

    private static Logger logger = LoggerFactory.getLogger(MessagePushServiceImpl.class);

    /**
     * 基线MQ泳道地址
     */
    @Value("${rocketmq.namesrvAddr}")
    private String basicNamesrvAddr;
    /**
     * 灰度MQ泳道地址
     */
    @Value("${rocketmq.canaryNamesrvAddr:null}")
    private String canaryNamesrvAddr;

    /**
     * 发送检查ack最大次数
     */
    @Value("${mq-message.ack-checking-limit:2}")
    private int messageAckCheckingLimit;

    @Resource
    private ProducerService producerService;

    @Resource
    private MessageRecordRepository messageRecordRepository;

    @Override
    public boolean pushMessage(MessageRecordPO messageRecordPO, MessageRecordStatusEnum oldMessageStatus) {
        return doPush(messageRecordPO, oldMessageStatus);
    }

    private boolean doPush(MessageRecordPO messageRecordPO, MessageRecordStatusEnum oldStatus) {
        if (!balance(messageRecordPO)) {
            return false;
        }

        BaseMQProducer producer = producerService.getProducerQuietly(messageRecordPO.getProducerKey());
        if (producer == null) {
            return false;
        }

        if (!messageRecordRepository.updateMessageStatus(messageRecordPO.getMessageId(), oldStatus, MessageRecordStatusEnum.SENDING)) {
            logger.info("修改消息状态 [PENDING -> SENDING] 失败，可能消息已被其他节点发送。messageId: {}", messageRecordPO.getMessageId());
            return false;
        }

        String msgKey = messageRecordPO.getMessageUuid();
        try {
            producer.send(
                    messageRecordPO.getMessageTopic(),
                    messageRecordPO.getMessageTag(),
                    msgKey,
                    messageRecordPO.getMessageJson(),
                    messageRecordPO.getProducerSource()
            );
            logger.info("消息发送成功. topic: {}, tags: {}, msgKey: {}, messageId: {}",
                    messageRecordPO.getMessageTopic(),
                    messageRecordPO.getMessageTag(),
                    msgKey,
                    messageRecordPO.getMessageId());
            if (!afterSendMessage(messageRecordPO)) {
                logger.warn("修改消息状态[sending -> sent]失败，可能消息状态已经发生改变。messageId:{}, messageData: {}", messageRecordPO.getMessageId(), messageRecordPO.getMessageJson());
            }
            return true;
        } catch (MQException e) {
            logger.error("发送消息失败, messageId: {}, topic:{}, tags: {}, msgKey: {}, retryCount: {}", messageRecordPO.getMessageId(), messageRecordPO.getMessageTopic(), messageRecordPO.getMessageTag(), msgKey, messageRecordPO.getRetryCount(), e);
            // 后置处理
            afterSendMessageFailed(messageRecordPO);
            return false;
        }
    }

    private boolean afterSendMessage(MessageRecordPO messageRecordPO) {
        //如果发送之前就是sending状态，说明是多次到MQ，在第一次发送成功后重置发送次数
        if (messageRecordPO.getMessageRecordStatus() == MessageRecordStatusEnum.SENDING
                || (messageRecordPO.getMessageRecordStatus() == MessageRecordStatusEnum.PENDING && messageRecordPO.getRetryCount() > 0)) {
            messageRecordRepository.updateMessageRetryCount(messageRecordPO.getMessageId(), 1);
        }
        return messageRecordRepository.updateMessageStatus(messageRecordPO.getMessageId(), null, MessageRecordStatusEnum.SENT);

    }

    /**
     * 泳道平衡
     *
     * @param messageRecordPO
     * @return
     */
    private boolean balance(MessageRecordPO messageRecordPO) {
        MQZone currentZone = getCurrentMQZone(messageRecordPO);
        MQZone recordZone = Objects.requireNonNullElse(messageRecordPO.getMqZone(), MQZone.BASIC);

        if (!Objects.equals(currentZone, recordZone)) {
            logger.info("消息发送失败: 当前泳道: {}, 消息泳道: {}, messageId: {}", currentZone, recordZone, messageRecordPO.getMessageId());
            return false;
        }
        return true;
    }

    /**
     * 消息发送失败处理
     *
     * @param record
     * @return
     */
    private void afterSendMessageFailed(MessageRecordPO record) {
        // 发送超过最大限制
        if (record.getRetryCount() >= messageAckCheckingLimit) {
            Boolean updateFlag = messageRecordRepository.updateMessageStatus(record.getMessageId(), null, MessageRecordStatusEnum.SENT_FAIL);
            logger.error("消息发送次数超过pending超时最大尝试次数 messageId: {},，修改状态为:{}, messageData: {}, 修改结果: {}", record.getMessageId(), MessageRecordStatusEnum.SENT_FAIL, record.getMessageJson(), updateFlag);
        }
    }


    /**
     * 获取当前mq泳道
     *
     * @param messageRecordPO
     * @return
     * @throws Exception
     */
    private MQZone getCurrentMQZone(MessageRecordPO messageRecordPO) {
        BaseMQProducer producer = producerService.getProducerQuietly(messageRecordPO.getProducerKey());
        if (producer == null) {
            return null;
        }

        String namesrvAddr = producer.getNamesrvAddr();
        if (basicNamesrvAddr.equals(namesrvAddr)) {
            return MQZone.BASIC;
        } else if (canaryNamesrvAddr.equals(namesrvAddr)) {
            return MQZone.CANARY;
        } else {
            return null;
        }
    }
}
